package com.crawler.search.utils;

import com.alibaba.fastjson.JSONObject;
import com.carrotsearch.hppc.cursors.ObjectObjectCursor;
import lombok.Data;
import org.apache.commons.lang3.StringUtils;
import org.elasticsearch.action.admin.indices.alias.IndicesAliasesRequest;
import org.elasticsearch.action.admin.indices.alias.get.GetAliasesRequest;
import org.elasticsearch.action.admin.indices.alias.get.GetAliasesResponse;
import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.action.search.SearchRequest;
import org.elasticsearch.action.search.SearchResponse;
import org.elasticsearch.action.search.SearchScrollRequest;
import org.elasticsearch.action.search.SearchType;
import org.elasticsearch.action.support.master.AcknowledgedResponse;
import org.elasticsearch.client.RequestOptions;
import org.elasticsearch.client.RestHighLevelClient;
import org.elasticsearch.cluster.metadata.AliasMetaData;
import org.elasticsearch.common.collect.ImmutableOpenMap;
import org.elasticsearch.common.unit.TimeValue;
import org.elasticsearch.search.SearchHit;
import org.elasticsearch.search.SearchHits;
import org.elasticsearch.search.builder.SearchSourceBuilder;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.io.IOException;
import java.util.List;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

/**
 * 数据导出
 *
 * @author zyh
 */
@Component
public class ReIndexUtils {

    @Autowired
    public RestHighLevelClient client;

    @Autowired
    public ESBulkProcessor bulkRequest;

    @Autowired
    public ESIndexUtils esIndexUtils;

	private static Logger logger = LoggerFactory.getLogger(ReIndexUtils.class);

    public static ExecutorService pool = new ThreadPoolExecutor(10, 100,
            0L, TimeUnit.MILLISECONDS,
            new LinkedBlockingQueue<Runnable>(1000));


    private static int i = 0;

    public void scollRebuild(String index,String path) {

        SearchRequest searchRequest = new SearchRequest(index)
                .searchType(SearchType.DEFAULT)
                .source(new SearchSourceBuilder().size(10))
                .scroll(new TimeValue(20000));

        SearchResponse searchResponse = null;
        try {
            searchResponse = client.search(searchRequest, RequestOptions.DEFAULT);

            parseSearchResponse(searchResponse,path);
            //获取总数量
            long totalCount = searchResponse.getHits().getTotalHits().value;
            int page = (int) totalCount / (10);//计算总页数,每次搜索数量为分片数*设置的size大小
			logger.info("index:{} ---- total:{} ---- page:{}",index,totalCount,page);
            for (int i = 0; i <= page; i++) {
                SearchScrollRequest scrollRequest = new SearchScrollRequest();
                scrollRequest.scrollId(searchResponse.getScrollId());
                scrollRequest.scroll(new TimeValue(2000000));
                //再次发送请求,并使用上次搜索结果的ScrollId
                searchResponse = client.scroll(scrollRequest, RequestOptions.DEFAULT);
                parseSearchResponse(searchResponse,path);
            }
        } catch (IOException e) {
			logger.error("scroll error msg : {}",e.toString());
        }

    }

    public void parseSearchResponse(SearchResponse searchResponse,String path) {
        SearchHits hits = searchResponse.getHits();
		logger.info("-----------begin------------");
		for (SearchHit searchHit : hits.getHits()) {
            try {
                i++;
                String index = searchHit.getIndex();
                JSONObject hitJO = JSONObject.parseObject(searchHit.getSourceAsString());
				Thread.sleep(2000);

				excuteThred(new ReIndex(hitJO.toJSONString(), index,path));
				logger.info("第" + i + "条数据:" + searchHit.getId() + "index:" + index);
			} catch (Exception e) {
				logger.error("scroll error msg : {}",e.toString());
			}
        }
		logger.info("-----------end------------");
	}

	@Data
    public class ReIndex implements Runnable {

        String data;

        String index;

        String path;


        public ReIndex(String data, String index,String path) {
            super();
            this.data = data;
            this.index = index;
            this.path = path;
        }

        @Override
		public void run() {
            try {
                //判断索引是否存在,不存在创建索引
                JSONObject object = JSONObject.parseObject(data);
                String indexAlias = index.replace("_v1", "_v2");
                if (!esIndexUtils.getAllIndices().contains(indexAlias)) {
                    esIndexUtils.checkAndCreateIndex(index.replace("_v1", ""), path, null);
                }
                JSONObject jsonObject = JSONObject.parseObject(data);
                String id = object.getString("id");
                if (StringUtils.isNotEmpty(id)) {
                    IndexRequest request = new IndexRequest(indexAlias).id(id).source(jsonObject);
                    bulkRequest.addIndexRequest(request);
                }
            } catch (Exception e) {
				logger.error("scroll error msg : {}",e.toString());
			}

        }
    }

    //线程执行
    public void excuteThred(Runnable run) {
        while (true) {
            try {
                pool.execute(run);
                break;
            } catch (Exception e) {
                e.printStackTrace();
                try {
                    Thread.sleep(100);
                } catch (InterruptedException e1) {
                    e1.printStackTrace();
                }
            }
        }
    }

}
